package com.bbx.flink.demo.window;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;



public class TumblingWindowDemo {


    public static void main (String [] args) throws Exception {
        //获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //连接数据源
        DataStreamSource<String> socketTextStream = env.socketTextStream("192.168.10.131", 10003);

        //数据转换
        socketTextStream
                .map(new MapFunction<String, Integer>() {
                    private static final long serialVersionUID = -3901220362086841385L;
                    @Override
                    public Integer map(String value) throws Exception {
                        return Integer.parseInt(value);
                    }
                })
                .keyBy(i->i%2)
                .window(TumblingProcessingTimeWindows.of(Time.minutes(1)))
//                .timeWindow(Time.minutes(3))
                .sum(0)
                .print();

        env.execute();




    }

}
